import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

public class Demo04_watermark {

    /**
     * 有序流
     * @param sensorDataStream
     */
    public static void watermark_ascend(DataStream<Demo02_sensorReading> sensorDataStream){
        sensorDataStream.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<Demo02_sensorReading>forMonotonousTimestamps()
                        //指定数据中哪个字段是时间戳，用于计算watermark
                        .withTimestampAssigner(
                            (element, recordTimestamp) -> element.getTimestamp()
                        )
        );
    }

    /**
     * 乱序流
     * @param sensorDataStream
     */
    public static void watermark_orderness(DataStream<Demo02_sensorReading> sensorDataStream){
        DataStream<Demo02_sensorReading> watermarkSensorStream = sensorDataStream.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<Demo02_sensorReading>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner(
                                (element, recordTimestamp) -> element.getTimestamp()
                        )
        );

        watermarkSensorStream
            .keyBy(Demo02_sensorReading::getSensorId)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .allowedLateness(Time.seconds(5))
            .sideOutputLateData(new OutputTag<Demo02_sensorReading>("lataData"));

        watermarkSensorStream.assignTimestampsAndWatermarks(new Demo04_myselfWatermarkStrategy());

    }


    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //接收String类型的原始数据
        DataStreamSource<String> dataStream = Demo02_source.source_socket(env,9999);
        //把String类型转成SensorReading类型
        DataStream<Demo02_sensorReading> sensorDataStream = Demo02_transformation.transformation_simple(dataStream);



        try {
            env.execute("window");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
